package cn.edu.thu.laud.thrift;
import java.util.Arrays;
import java.util.List;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.AbstractCassandraDaemon;
import org.apache.cassandra.thrift.AuthenticationException;
import org.apache.cassandra.thrift.AuthorizationException;
import org.apache.cassandra.thrift.CassandraDaemon;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class LaUDDaemon  extends CassandraDaemon{
	public static LaUDDaemon instance;
	static
	{
		AbstractCassandraDaemon.initLog4j();
	}
	static Logger logger = LoggerFactory.getLogger(CassandraDaemon.class);//hxd:customized logger
	final static String SYNC = "sync";
	final static String ASYNC = "async";
	final static String HSHA = "hsha";
	public final static List<String> rpc_server_types = Arrays.asList(SYNC, ASYNC, HSHA);
	private LaUDServer server;//thrift server
	private LaUDDataServer dataServer;//hxd:file data server
	private String[] args;
	
	//a client who  controls the user authority..
	LaUDAdministrator daemonClient;
	
	
	public LaUDCassandraServer getCassandraServer(){
		return this.server.getLaUDCassandraServer();
	}
	protected void startServer()
	{
		//try {
		//listenAddr= InetAddress.getByName(LaUDLogUtils.parser.getElementString("ServerIP",listenAddr.getHostAddress()));
		//listenPort=Integer.parseInt(LaUDLogUtils.parser.getElementString("ServerPort",listenPort));
		//			} catch (UnknownHostException e1) {
		//				e1.printStackTrace();
		//			}
		if (server == null)
		{
			server = new LaUDServer(listenAddr, listenPort,args);
			server.start();
		}
		if(dataServer==null){
			try {
				dataServer=new LaUDDataServer(listenAddr,Integer.parseInt(LaUDLogUtils.parser.getElementString("DataServerPort",listenPort+1)));
				dataServer.start();
			} catch (Exception e) {
				e.printStackTrace();
				stop();
			}
		}
		
		
	}

	protected void stopServer()
	{
		logger.warn("laudDaemon will be closed..................");
		//hxd: add close data server
		if(dataServer!=null){
			dataServer.stopServer();
			dataServer=null;
		}

		if (server != null)
		{
			server.stopServer();
			try
			{
				server.join();
			}
			catch (InterruptedException e)
			{
				logger.error("Interrupted while waiting thrift server to stop", e);
			}
			server = null;
		}
	}

	public static void stop(String[] args)
	{
		instance.stopServer();
		instance.deactivate();
		System.exit(1);
	}

	public void initDaemonAdministrator() throws AuthenticationException, AuthorizationException, TException, InvalidRequestException, SchemaDisagreementException, UnavailableException, TimedOutException {
		daemonClient=new LaUDAdministrator(DatabaseDescriptor.getRpcAddress().getHostAddress(), 9170, 0,"root");
		daemonClient.createSystemKeyspaces();
		daemonClient.loadUsers();
	}
	public void quitAdministrator(){
		if(daemonClient!=null&&daemonClient.isOpen()){
			daemonClient.close();
		}
	}
	public static void main(String[] args)
	{
		try{
			
			instance = new LaUDDaemon();
			instance.args=args;
			instance.activate();
			//暂停权限管理，因为底层存储禁用了super column
			//instance.initDaemonAdministrator();
		}catch(Exception e){
			e.printStackTrace();
			logger.error(e.getMessage());
			stop(args);
		}
	}

}
